package datastream.api.operator.one;

import datastream.api.operator.one.source.user.UserSource;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import datastream.api.operator.one.pojo.user.InputModel;
import datastream.api.operator.one.pojo.user.OutputModel;

public class Map {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<InputModel> source = env.addSource(new UserSource());

        SingleOutputStreamOperator<OutputModel> map = source.map(new MapFunction<InputModel, OutputModel>() {
            @Override
            public OutputModel map(InputModel value) throws Exception {
                String ageRange = "UNKNOWN";
                int age = value.getAge();

                if (age >= 0 && age < 18) {
                    ageRange = "0-18";
                } else if (age < 50) {
                    ageRange = "18-50";
                } else {
                    ageRange = "50+";
                }

                return OutputModel.builder()
                        .username(value.getUsername())
                        .ageRange(ageRange)
                        .build();
            }
        });

        DataStreamSink<OutputModel> sink = map.print();

        env.execute();
    }
}
